package com.jacquesh.sink;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

public class RedisSinkDemo {
    public static void main(String[] args) throws Exception {
        /***
         * 配置redis 基本配置信息
         */
        FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder()
                .setHost("192.168.1.27")
                .setPort(6379)
                .setPassword("123456")
                .setDatabase(1)
                .build();

        StreamExecutionEnvironment environment=
                StreamExecutionEnvironment.getExecutionEnvironment();


        /**
         * 监听socket stream
         */
        DataStreamSource<String> stream= environment
                .socketTextStream("192.168.88.170",8888);
        /**
         * 加入redis Sink  第一各参数为redis 配置信息
         * 第二个参数为 redis的item的格式信息 key 设置 内容设置
         */
        stream.addSink(new RedisSink<String>(config, new RedisMapper<String>() {
            @Override

            public RedisCommandDescription getCommandDescription() {
                /**
                 * 设置存储类型   第二个参数为key
                 */
                return new RedisCommandDescription(RedisCommand.LPUSH,"wc");
            }
            @Override
            public String getKeyFromData(String s) {
                /**
                 * 内容设置
                 */
                return s;
            }

            @Override
            public String getValueFromData(String s) {
                /**
                 * 内容设置
                 */
                return s;
            }
        }));
        environment.execute("redis demo");
    }

}
